package com.wl.cloud.datax.common.utils;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.List;

/**
 * @author: wanglin
 * @date: 2023-08-17 周四
 * @Version: 1.0
 * @Description:
 */
@SuppressWarnings("all")
@Slf4j
public class CustomHbaseUtils {

    /**
     * 初始化连接
     *
     * @param ip
     * @param clientPort
     * @return
     */
    public static Connection getConnection(String ip, String clientPort) {
        //配置类，通过连接zookeeper连接hbase，后面跟主机名，不要跟ip地址，自己配置windows映射关系
        Configuration conf = new Configuration();
        //" IP_ADDRESS " 处，输入对应的虚机的IP地址，或者输入对应的虚机的主机名
        conf.set("hbase.zookeeper.quorum", ip);
        //" 端口号 " 处，输入对应的端口号
        conf.set("hbase.zookeeper.property.clientPort", clientPort);
        //连接hbase
        Connection conn = null;
        try {
            conn = ConnectionFactory.createConnection(conf);
        } catch (IOException e) {
            e.printStackTrace();
            log.error("初始hbase连接异常：{}", e.getMessage(), e);
        }
        log.info("初始化Hbase连接成功");
        return conn;
    }

    /**
     * 关闭连接
     *
     * @param admin
     * @param hbaseConnection
     */
    public static void close(Connection hbaseConnection) {
        try {
            // 一定注意，这里这个连接不能关闭，否则后面就都没有了，让spring自己管理，我就呗这里坑了半天
            if (null != hbaseConnection) {
                hbaseConnection.close();
                log.info("close Hbase Connection");
            }
        } catch (IOException e) {
            e.printStackTrace();
            log.error("Hbase连接关闭失败");
        }
    }

    /**
     * 关闭连接
     *
     * @param admin
     * @param hbaseConnection
     */
    public static void close(Admin admin) {
        try {
            if (null != admin) {
                admin.close();
                log.info("close Hbase Admin");
            }
        } catch (IOException e) {
            e.printStackTrace();
            log.error("Hbase admin closed failed");
        }
    }

    /**
     * 关闭连接
     *
     * @param admin
     * @param hbaseConnection
     */
    public static void close(Admin admin, Connection hbaseConnection) {
        try {
            if (null != admin) {
                admin.close();
            }
            if (null != hbaseConnection) {
                hbaseConnection.close();
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 判断表是否存在
     *
     * @param admin     hbase连接admin
     * @param tableName 表对象
     * @return 布尔值，true是（存在），false否（不存在）
     * @throws IOException
     */
    public static boolean checkTableExist(Admin admin, String tableName) {
        if (admin == null || StringUtils.isBlank(tableName)) {
            return false;
        }
        TableName tbName = TableName.valueOf(tableName);
        if (tbName == null) {
            return false;
        }
        try {
            return admin.tableExists(tbName);
        } catch (IOException e) {
            e.printStackTrace();
            log.error("检查表：" + tableName + "失败");
            return false;
        }
    }

    /**
     * 判断表是否存在
     *
     * @param admin     hbase连接admin
     * @param tableName 表对象
     * @return 布尔值，true是，false否
     * @throws IOException
     */
    public static boolean checkTableExist(Admin admin, TableName tableName) throws IOException {
        if (admin == null || tableName == null) {
            return false;
        }
        return admin.tableExists(tableName);
    }


    /**
     * 创建一个表
     *
     * @throws IOException
     */
    public static boolean createTable(Admin admin, String tbName) {
        if (admin == null || StringUtils.isBlank(tbName)) {
            return false;
        }
        //4.创建表的相关信息,表名
        HTableDescriptor tableDescriptor = new HTableDescriptor(TableName.valueOf(tbName));
        //5.添加列族信息
        tableDescriptor.addFamily(new HColumnDescriptor("info"));
//        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("name"), Bytes.toBytes("zhangsan"));
//        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("gender"), Bytes.toBytes("male"));
//        tableDescriptor.addFamily(new HColumnDescriptor("id"));
//        tableDescriptor.addFamily(new HColumnDescriptor("flag"));
        //6.调用创建表的方法进行建表操作
        try {
            admin.createTable(tableDescriptor);
        } catch (IOException e) {
            e.printStackTrace();
            log.error("Hbase创建表：{}失败，错误信息为：{}", tbName, e.getMessage());
            return false;
        }
        log.info("Hbase创建表：{}成功", tbName);
        return true;
    }

    /**
     * 新增
     *
     * @throws IOException
     */
    public static void putDataTable() throws IOException {
        //1.获取hbase连接，配置
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "192.168.204.100");
        conf.set("hbase.zookeeper.property.clientPort", "2181");
        //2.创建连接
        Connection conn = ConnectionFactory.createConnection(conf);
        //3.获取table
        Table student = conn.getTable(TableName.valueOf("student100"));
        //4.表中添加数据rowkey
        Put put = new Put(Bytes.toBytes("1001"));
        //5.添加列，列族名：列名 值，如：info:name zhangsan
        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("name"), Bytes.toBytes("zhangsan"));
        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("gender"), Bytes.toBytes("male"));
        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("age"), Bytes.toBytes("18"));
        //6.插入数据
        student.put(put);
        //7.关闭连接
        conn.close();
    }

    /**
     * 读取数据
     *
     * @throws IOException
     */
    public static void getDataFromTable() throws IOException {
        //1.获取hbase连接，配置
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "192.168.204.100");
        conf.set("hbase.zookeeper.property.clientPort", "2181");
        //2.创建连接
        Connection conn = ConnectionFactory.createConnection(conf);
        //3.获取table
        Table student = conn.getTable(TableName.valueOf("student100"));
        //4.读取数据，Get
        Get get = new Get(Bytes.toBytes("1001"));
        //5.获取结果
        Result result = student.get(get);

        //6.遍历
        Cell[] cells = result.rawCells();
        for (Cell cell : cells) {
            System.out.println("rowkey:" + Bytes.toString(CellUtil.cloneRow(cell)));
            System.out.println("列族:" + Bytes.toString(CellUtil.cloneFamily(cell)));
            System.out.println("列名:" + Bytes.toString(CellUtil.cloneQualifier(cell)));
            System.out.println("value:" + Bytes.toString(CellUtil.cloneValue(cell)));
            System.out.println("-----------------");
        }
        conn.close();
    }

    /**
     * 读取数据
     *
     * @throws IOException
     */
    public static void getDataByList(Table table, List<Get> keyList) throws IOException {
        Result[] results = table.get(keyList);
        for (int i = 0; i < results.length; i++) {
            Result result = results[i];
            boolean b = result.isEmpty();
            Cell[] cells = results[i].rawCells();
            //6.遍历
            for (Cell cell : cells) {
                System.out.println("rowkey:" + Bytes.toString(CellUtil.cloneRow(cell)));
                System.out.println("列族:" + Bytes.toString(CellUtil.cloneFamily(cell)));
                System.out.println("列名:" + Bytes.toString(CellUtil.cloneQualifier(cell)));
                System.out.println("value:" + Bytes.toString(CellUtil.cloneValue(cell)));
                System.out.println("-----------------");
            }
        }
    }

    /**
     * 删除
     *
     * @param tableName
     * @throws IOException
     */
    public static boolean dropTable(Admin admin, String tableName) {
        try {
            //4.禁用表
            admin.disableTable(TableName.valueOf(tableName));
            //5.删除表
            admin.deleteTable(TableName.valueOf(tableName));
            log.info("删除表：" + tableName);
        } catch (IOException e) {
            e.printStackTrace();
            log.error("删除表：" + tableName + "失败");
            return false;
        }
        return true;
    }
}
